ChatWorkflowRunner: multi-turn chat as dynamic workflow

Context

Multiple consumers need multi-turn conversation state:

lightspeed-stack — `/query` chatbot with conversation cache (SQLite/PostgreSQL)
cloud-agents CLI sessions — `cli_session.py` manages bi-directional messaging
Workflow steps — context chaining via `{{ steps.X.output.Y }}` (different mechanism, same need)

Currently each consumer hand-rolls conversation management. A shared ABC would unify the interface while allowing different storage backends.

Proposal

class ConversationStore(ABC):
    """Abstract interface for turn-level conversation storage."""

    @abstractmethod
    async def load_context(self, conversation_id: str) -> dict[str, Any]:
        """Load accumulated context for a conversation."""

    @abstractmethod
    async def save_turn(
        self, conversation_id: str, prompt: str, result: StepResult
    ) -> None:
        """Save a completed turn to the conversation."""

    @abstractmethod
    async def list_turns(self, conversation_id: str) -> list[dict[str, Any]]:
        """List turns in a conversation."""

How it fits

The conversation store is NOT inside the executor. The pattern is:

# Caller (endpoint or orchestrator) manages conversation state:
context = await store.load_context(conversation_id)
step_input = StepInput(prompt=prompt, context=context, ...)
result = await executor.run(step_input)
await store.save_turn(conversation_id, prompt, result)

The executor stays stateless — it receives context and returns results. The caller manages the conversation lifecycle.

Implementations

Consumer	Backend
lightspeed-stack	Existing `Cache` (SQLite/PostgreSQL)
cloud-agents CLI	In-memory (session-scoped)
cloud-agents workflows	`RunStateStore` (already persists step results)

Integration with StepInput

Add `conversation_id` to `StepInput.metadata` (see #144) rather than as a top-level field. The conversation store reads it from metadata in the middleware layer.

Triage

Summary: Shared ConversationStore ABC for multi-turn agent sessions across lightspeed-stack, cloud-agents CLI, and workflow steps.

Affected components:

src/cloud_agents/workflow/executor/step/direct.py — DirectExecutor would use ConversationStore for multi-turn context
src/cloud_agents/workflow/executor/step/base.py — StepInput may need conversation_id field
New: src/cloud_agents/storage/conversation_store.py — ABC + PostgreSQL implementation

Related issues/PRs:

Design: Embed cloud agents harness into lightspeed-stack as optional feature #125 — embed cloud agents into lightspeed-stack (shares conversation state)
StepExecutor.run_stream(): streaming support for all spawn modes #135 — streaming support (streaming + multi-turn = chatbot UX)
PR T37: Secret redaction in logs and error responses #7 (lightspeed-stack) — /query/direct already has context param

Suggested labels: enhancement

🤖 Triaged by beesarmy via Claude Code

Alternative: multi-turn chat as a dynamic workflow

Instead of a separate ConversationStore ABC, treat a multi-turn chat as a workflow with dynamically appended steps. Each user message is one step executed via DirectExecutor, with prior turn outputs flowing as context.

Why this works

The existing infrastructure already supports it:

RunStateStore accumulates step results per workflow — that IS conversation history
TranscriptStore records execution details per step — that IS per-turn audit
StepInput.context passes all prior step outputs — that IS conversation context for the LLM
workflow_id becomes conversation_id

What changes

Instead of building a new store, we need a ConversationalExecutor — a thin wrapper that accepts one message at a time and accumulates in RunStateStore:

class ConversationalExecutor:
    async def send_message(self, conversation_id: str, prompt: str) -> StepResult:
        """Execute one turn, with all prior turns as context."""
        prior = await self.store.get(conversation_id)
        step_input = StepInput(
            prompt=prompt,
            context=prior.steps,  # all prior turns
            tools=self.config.tools,
            mcp_servers=self.config.mcp_servers,
            provider=self.config.provider,
        )
        result = await self.executor.run(step_input)
        await self.store.update_step(conversation_id, f"turn-{n}", result)
        return result

    async def send_message_stream(self, conversation_id: str, prompt: str) -> AsyncIterator[StreamEvent]:
        """Same, but streaming."""
        # ... same pattern with executor.run_stream()

Two modes, one engine

	Predefined workflow	Conversational workflow
Steps	Defined in YAML, finite	Added per user message, open-ended
Context	Prior step outputs	Same — prior turn outputs
Tools/MCP/Skills	Per-step from YAML	Per-conversation config
State	RunStateStore	Same RunStateStore
Ends when	Last step completes	User ends session

What this means for lightspeed-stack integration

The /query endpoint becomes:

load/create conversation → executor.send_message(conversation_id, prompt) → stream SSE → done

No separate conversation cache needed. RunStateStore (PostgreSQL, already shared with LCS) is the store.

Recommendation

Keep this issue open but reframe it: instead of ConversationStore ABC, implement ConversationalExecutor that reuses RunStateStore + StepExecutor. The original ConversationStore proposal becomes a fallback if the RunStateStore approach proves too heavyweight for simple chatbot use cases.

Refined alternative: conversation history as a view on transcripts

Building on the "chat as dynamic workflow" alternative, conversation history is not a separate store — it's a projected view on transcript data.

The data model

Transcript (full detail per step/turn)
├── LLM request: prompt sent to model
├── LLM response: model output
├── Tool call: name + args
├── Tool result: output
├── MCP call: server + tool + result
└── Timing, tokens, cost

Conversation history (filtered view)
├── Turn 1: user prompt → agent response
├── Turn 2: user prompt → agent response
└── Turn 3: user prompt → agent response

Three abstractions eliminated

Original proposal	Becomes
`ConversationStore`	Query on `TranscriptStore`
`conversation_id`	= `workflow_id`
Conversation history	= Transcript filtered to prompts + outputs

How it works

Each conversation is one workflow. Each turn is one step with a detailed transcript.

# Building LLM context for next turn
prior_turns = await transcript_store.list_steps(conversation_id)
messages = []
for turn in prior_turns:
    messages.append({"role": "user", "content": turn.prompt})
    messages.append({"role": "assistant", "content": turn.output})

The TranscriptStore already has:

save(workflow_id, step_name, events, cost, tokens, duration) — stores one turn
list_steps(workflow_id) — lists all turns in a conversation
get(workflow_id, step_name) — gets one turn's detail
PostgreSQL backend with configurable retention

What TranscriptStore needs

To serve as conversation history, it needs:

Store prompt alongside transcript events — currently only stores events (JSONB), not the original prompt. Add prompt and output columns, or include them as transcript event types.
Efficient conversation loading — list_steps() returns step names only. Need a query that returns prompt + output for all turns without loading full event logs.
Message-history format — a helper that converts stored turns to pydantic-ai message_history format for Agent.run(message_history=...).

Recommendation

Close this issue in favor of:

Extend TranscriptStore schema to include prompt/output columns (StepMiddleware protocol for cross-cutting executor concerns #144 middleware stores these)
Add load_conversation_context(workflow_id) query method
ConversationalExecutor uses TranscriptStore directly — no new store ABC needed

Dependency: #146 (identity and association model) provides session_id, parent_workflow_id, and TranscriptStore schema extensions (prompt/output columns) that the conversational executor needs. #146 should land first.

Opus review revisions

Prototype message_history first

The most critical design question is whether pydantic-ai Agent.run(message_history=...) works for multi-turn conversation. Current StepInput.context flattens prior turns to text — this loses the structured role/content format LLMs need.

Before building ConversationalExecutor, prototype:

# Load prior turns from TranscriptStore
turns = await transcript_store.load_recent_turns(conversation_id, limit=20)

# Convert to pydantic-ai ModelMessage objects
from pydantic_ai.messages import ModelRequest, ModelResponse
message_history = []
for turn in turns:
    message_history.append(ModelRequest.user_text_prompt(turn.prompt))
    # Need to reconstruct ModelResponse from stored output
    ...

# Feed to Agent
result = await agent.run(new_prompt, message_history=message_history)

If ModelMessage objects can be serialized/deserialized to/from the TranscriptStore schema, the architecture works. If not, we need to store the raw pydantic-ai message format.

Context window management

Add max_context_turns config (default 20) + load_recent_turns() query. For longer conversations, consider lightspeed-stack's conversation_compaction.py which summarizes older turns.

Store orchestration

ConversationalExecutor explicitly uses BOTH stores:

RunStateStore — create/update workflow lifecycle (status, current turn)
TranscriptStore — save/load turn content (prompt, output, events)

class ConversationalExecutor:
    def __init__(self, executor, run_store, transcript_store, config):
        ...

    async def send_message(self, conversation_id, prompt):
        # Load recent turns from transcript_store
        history = await self.transcript_store.load_recent_turns(conversation_id)
        # Build message_history for pydantic-ai
        ...
        # Execute via step executor
        result = await self.executor.run(step_input)
        # Save turn to transcript_store
        await self.transcript_store.save(conversation_id, turn_name, ...)
        # Update workflow state in run_store
        await self.run_store.update_step(conversation_id, turn_name, ...)

Concurrent conversations

Works — each conversation gets a unique workflow_id. Same user can have multiple conversations simultaneously. No conflict.

GDPR deletion

RunStateStore.delete() + TranscriptStore.delete_workflow() already exist. Need a /v1/conversations/{id} DELETE endpoint. Track as follow-up.

message_history prototype: VALIDATED

Prototyped pydantic-ai's Agent.run(message_history=...) — the architecture is viable.

What was proven

Step	Result
Reconstruct `ModelRequest`/`ModelResponse` from stored data	Works
Serialize via `TypeAdapter(list[ModelMessage]).dump_json()`	Works — ~850 bytes/turn
Deserialize back to typed objects	Round-trips perfectly
`Agent.run(message_history=restored)` accepts the type	Works
`result.all_messages()` returns prior + new messages	Works by design

Design decision: store serialized ModelMessage JSON (Option B)

Store the full pydantic-ai message objects in TranscriptStore, not just prompt/output text. This preserves tool call details — the LLM sees that it previously called kubectl_get and got a specific result, not just a text summary.

Schema change:

ALTER TABLE step_transcripts ADD COLUMN
    message_history JSONB  -- serialized list[ModelMessage] via TypeAdapter
;

ConversationalExecutor flow:

from pydantic import TypeAdapter
from pydantic_ai.messages import ModelMessage

_MESSAGE_TA = TypeAdapter(list[ModelMessage])

class ConversationalExecutor:
    async def send_message(self, conversation_id, prompt):
        # Load prior message history
        stored = await self.transcript_store.load_message_history(conversation_id)
        prior_messages = _MESSAGE_TA.validate_json(stored) if stored else []
        
        # Execute with full conversation context
        agent = Agent(model, tools=tools, ...)
        result = await agent.run(prompt, message_history=prior_messages)
        
        # Store ALL messages (prior + new turn) for next time
        all_messages = result.all_messages()
        serialized = _MESSAGE_TA.dump_json(all_messages)
        await self.transcript_store.save_turn(conversation_id, turn_name, serialized)
        
        return result

Why Option B over prompt/output text:

LLM sees prior tool calls and results (not just text summaries)
Structured role/content format (not text concatenation)
pydantic-ai handles serialization natively
~850 bytes per turn — manageable even for 100+ turns
result.all_messages() returns the complete history including the new turn — store once, load once

Context window management:

max_context_turns config limits how many prior turns to load
For long conversations, load only the last N turns' message_history
Future: compaction middleware summarizes older turns (lightspeed-stack's conversation_compaction.py pattern)

Design revision: framework-agnostic message format

Problem

Storing pydantic-ai ModelMessage JSON directly couples our persistence layer to pydantic-ai's internal types. If we swap frameworks, the stored data becomes unreadable.

Fix: our own message format + adapter at the boundary

@dataclass
class ConversationMessage:
    """Framework-agnostic conversation message.
    
    This is OUR format — stable across framework changes.
    """
    role: Literal["user", "assistant", "tool_call", "tool_result"]
    content: str
    timestamp: datetime
    metadata: dict[str, Any] = field(default_factory=dict)
    # metadata carries: tool_name, tool_args, model, tokens, etc.

Adapter pattern (same as ToolDefinition → pydantic_ai.Tool)

Storage layer:   ConversationMessage[]  (our format, stable)
                        ↕ adapter
Runtime layer:   ModelMessage[]         (pydantic-ai, swappable)

# Save: framework messages → our format
def from_model_messages(messages: list[ModelMessage]) -> list[ConversationMessage]:
    result = []
    for msg in messages:
        if msg.kind == "request":
            for part in msg.parts:
                if hasattr(part, "content"):  # UserPromptPart
                    result.append(ConversationMessage(role="user", content=part.content, ...))
        elif msg.kind == "response":
            for part in msg.parts:
                if hasattr(part, "content"):  # TextPart
                    result.append(ConversationMessage(role="assistant", content=part.content, ...))
                elif hasattr(part, "tool_name"):  # ToolCallPart
                    result.append(ConversationMessage(
                        role="tool_call", content="",
                        metadata={"tool_name": part.tool_name, "args": part.args},
                    ))
    return result

# Load: our format → framework messages
def to_model_messages(messages: list[ConversationMessage]) -> list[ModelMessage]:
    """Convert to pydantic-ai ModelMessage for Agent.run(message_history=...)"""
    ...

# If we swap to another framework — only the adapter changes
def to_langchain_messages(messages: list[ConversationMessage]) -> list:
    ...

What this means for storage

ALTER TABLE step_transcripts ADD COLUMN
    trace_id TEXT,
    messages JSONB  -- serialized list[ConversationMessage], NOT ModelMessage
;

The JSONB stores our ConversationMessage format. Loading for any framework goes through the adapter. The prototype validation still holds — we just add a conversion layer.

Consistency with existing patterns

Our abstraction	Framework type	Where
`ToolDefinition`	`pydantic_ai.Tool`	`tools.py` — `get_tools()` converts at runtime
`ConversationMessage`	`ModelMessage`	adapter — converts at load/save
`StepExecutor`	`pydantic_ai.Agent`	executor — wraps at execution

Same principle throughout: our contract is stable, framework is pluggable.

Naming revision: ConversationalExecutor → AgentSession

Why rename

"ConversationalExecutor" implies it's a StepExecutor subclass (like DirectExecutor, SubprocessExecutor). It's not — it's a higher-level wrapper that manages turn state and delegates to a real executor.

What it actually does

Load prior turns from TranscriptStore
Convert ConversationMessage[] → pydantic-ai ModelMessage[] via adapter
Build StepInput with conversation context
Delegate to StepExecutor.run() or run_stream() for the actual LLM call
Convert new messages → ConversationMessage[] and save to TranscriptStore
Update workflow lifecycle in RunStateStore

It manages a session, not an execution.

Interface

class AgentSession:
    """Multi-turn agent session backed by TranscriptStore."""

    def __init__(
        self,
        executor: StepExecutor,
        transcript_store: TranscriptStore,
        run_store: RunStateStore,
        config: AgentSessionConfig,
    ): ...

    async def send_message(
        self, conversation_id: str, prompt: str
    ) -> StepResult:
        """Execute one turn with all prior turns as context."""

    async def send_message_stream(
        self, conversation_id: str, prompt: str
    ) -> AsyncIterator[StreamEvent]:
        """Same, but streaming."""

    async def get_history(
        self, conversation_id: str, limit: int = 20
    ) -> list[ConversationMessage]:
        """Load recent conversation history."""

    async def create(self, user_id: str) -> str:
        """Create a new conversation. Returns conversation_id."""

    async def close(self, conversation_id: str) -> None:
        """Mark conversation as completed."""

How lightspeed-stack uses it

# In /query endpoint
session = AgentSession(executor, transcript_store, run_store, config)

# New conversation
conv_id = await session.create(user_id=caller.user_id)

# Each user message
async for event in session.send_message_stream(conv_id, user_prompt):
    yield sse_format(event)

Naming consistency

Concept	Name	Role
`StepExecutor`	DirectExecutor, SubprocessExecutor, SandboxExecutor	Executes one step
`AgentSession`	—	Manages multi-turn conversation, delegates to StepExecutor
`LocalExecutor`	—	Orchestrates predefined workflows, delegates to StepExecutor

AgentSession is to conversations what LocalExecutor is to predefined workflows — a higher-level orchestrator that uses StepExecutor under the hood.

Final naming: AgentSession → ChatWorkflowRunner

Naming hierarchy

WorkflowRunner              — predefined YAML steps (pydantic-graph)
TemporalWorkflowRunner      — predefined steps with crash recovery (Temporal)
ChatWorkflowRunner           — dynamic steps from user messages
    └── all three delegate to StepExecutor for individual steps

"ChatWorkflowRunner" because:

Peer naming with WorkflowRunner/TemporalWorkflowRunner
"Chat" says what drives it (user messages)
"WorkflowRunner" says what it is (an orchestrator over StepExecutor)
Reinforces "chat IS a workflow" — it's in the name

Updated interface

class ChatWorkflowRunner:
    """Multi-turn chat as a dynamic workflow."""

    def __init__(
        self,
        executor: StepExecutor,
        transcript_store: TranscriptStore,
        run_store: RunStateStore,
        config: ChatWorkflowConfig,
    ): ...

    async def create(self, user_id: str) -> str:
        """Create a new conversation workflow. Returns conversation_id."""

    async def send_message(self, conversation_id: str, prompt: str) -> StepResult:
        """Execute one turn with prior turns as message_history."""

    async def send_message_stream(
        self, conversation_id: str, prompt: str
    ) -> AsyncIterator[StreamEvent]:
        """Same, but streaming."""

    async def get_history(
        self, conversation_id: str, limit: int = 20
    ) -> list[ConversationMessage]:
        """Load recent conversation history."""

    async def close(self, conversation_id: str) -> None:
        """Mark conversation workflow as completed."""

How lightspeed-stack uses it

runner = ChatWorkflowRunner(executor, transcript_store, run_store, config)
conv_id = await runner.create(user_id=caller.user_id)

async for event in runner.send_message_stream(conv_id, user_prompt):
    yield sse_format(event)

All previous references to "ConversationalExecutor" and "AgentSession" in this issue are superseded by ChatWorkflowRunner.

Review

Closing ConversationStore as an ABC is the right call inside this repo. Do not reopen it. ChatWorkflowRunner + transcript view is the cloud-agents design.

It is the wrong integration story for lightspeed-stack, and that needs to be explicit here so #125 does not swallow LCS's cache.

Do

Prototype is done; implementation can proceed on: framework-agnostic ConversationMessage, max_context_turns, load_recent_turns(), GDPR delete endpoint.
Store serialized our messages, not raw pydantic-ai ModelMessage JSON (the later revision). Adapter at the Agent boundary.
ChatWorkflowRunner uses both stores: RunStateStore for lifecycle, TranscriptStore for turns. Good.

Do not

Equate conversation_id with workflow_id at the LCS boundary. LCS already has conversation_cache (SQLite/Postgres), compaction (conversation_compaction.py), list/delete conversation APIs, and Action.QUERY_OTHERS_CONVERSATIONS. Mapping 1:1 couples chat retention to workflow-run retention and bypasses compaction.
Put conversation load/save inside DirectExecutor. The later "caller owns the loop" pattern is correct; keep it.
Treat this as a Phase 1 Design: Embed cloud agents harness into lightspeed-stack as optional feature #125 dependency. Mounting /v1/workflows does not need a chat runner. LCS /query parity is a later adapter: LCS cache → ConversationMessage[] → ChatWorkflowRunner.send_message → map SSE to format_stream_data. That adapter lives in LCS, not here.

Missing from the issue body (update it)

The title still says ConversationStore ABC. Body + comments disagree. Rewrite the title/checklist to ChatWorkflowRunner or people will implement the ABC from the first revision.

Also specify: concurrent turns on the same conversation_id (last-write-wins vs 409), and that message_history growth must be bounded before this is used as /query. max_context_turns=20 is a start; LCS compaction is the actual long-thread strategy and should stay in LCS.

Acknowledged — title fixed, scope clarified

Title updated to "ChatWorkflowRunner: multi-turn chat as dynamic workflow". ConversationStore ABC is dead.

Accepted:

ConversationMessage format, max_context_turns, load_recent_turns()
ChatWorkflowRunner uses both stores (RunStateStore lifecycle, TranscriptStore turns)
Adapter at Agent boundary, not raw ModelMessage storage
GDPR delete endpoint
Concurrent turns on same conversation_id: define behavior (last-write-wins vs 409)

NOT a Phase 1 #125 dependency. Embedding /v1/workflows doesn't need a chat runner. LCS /query parity is a later adapter that lives in LCS, not here.

Do not equate conversation_id = workflow_id at the LCS boundary — LCS has its own conversation cache with compaction, list/delete APIs, and RBAC.

Clarification: ChatWorkflowRunner works in ALL modes

ChatWorkflowRunner is available in both standalone and embedded deployments. The unification of chat and workflows IS the design goal — not something to avoid in embedded mode.

When embedded in lightspeed-stack:

ChatWorkflowRunner is an option teams can use for /query
It brings the unified model: chat = workflow, same executor stack, same tools/MCP/skills
LCS's existing conversation cache can be gradually migrated or adapted — that's an LCS migration concern (Design: Embed cloud agents harness into lightspeed-stack as optional feature #125), not a cloud-agents architecture concern

Cloud-agents provides the capability. LCS decides how to adopt it.

The earlier comment about "NOT a Phase 1 #125 dependency" still stands — mounting /v1/workflows doesn't require ChatWorkflowRunner. But ChatWorkflowRunner is available for teams that want unified chat+workflow from day one.

Merge pull request #152 from jameswnl/issue-145-chat-workflow-runner

Merge pull request #153 from jameswnl/issue-145-remaining-fixes

Merge pull request #154 from jameswnl/issue-145-final-fixes

Merge pull request #157 from jameswnl/fix/chat-correctness

ChatWorkflowRunner: multi-turn chat as dynamic workflow #145

Description

Context

Proposal

How it fits

Implementations

Integration with StepInput

Related

Activity

beesarmy commented on Aug 23, 2026

Triage

jameswnl commented on Aug 23, 2026

Alternative: multi-turn chat as a dynamic workflow

Why this works

What changes

Two modes, one engine

What this means for lightspeed-stack integration

Recommendation

jameswnl commented on Aug 23, 2026

Refined alternative: conversation history as a view on transcripts

The data model

Three abstractions eliminated

How it works

What TranscriptStore needs

Recommendation

jameswnl commented on Aug 23, 2026

jameswnl commented on Aug 23, 2026

Opus review revisions

Prototype message_history first

Context window management

Store orchestration

Concurrent conversations

GDPR deletion

jameswnl commented on Aug 23, 2026

message_history prototype: VALIDATED

What was proven

Design decision: store serialized ModelMessage JSON (Option B)

jameswnl commented on Aug 23, 2026

Design revision: framework-agnostic message format

Problem

Fix: our own message format + adapter at the boundary

Adapter pattern (same as ToolDefinition → pydantic_ai.Tool)

What this means for storage

Consistency with existing patterns

jameswnl commented on Aug 23, 2026

Naming revision: ConversationalExecutor → AgentSession

Why rename

What it actually does

Interface

How lightspeed-stack uses it

Naming consistency

jameswnl commented on Aug 23, 2026

Final naming: AgentSession → ChatWorkflowRunner

Naming hierarchy

Updated interface

How lightspeed-stack uses it

jameswnl commented on Aug 23, 2026

Review

jameswnl commented on Aug 23, 2026

Acknowledged — title fixed, scope clarified

jameswnl commented on Aug 23, 2026

Clarification: ChatWorkflowRunner works in ALL modes

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions