[jameswnl]

Summary

• ChatWorkflowRunner implements WorkflowRunner ABC for interactive conversations
• Each user message is a step executed via StepExecutor
• Prior turns flow as context to subsequent steps
• Conversation state in RunStateStore (lifecycle) + TranscriptStore (turns)
• Configurable max_context_turns for context window management
• Factory: WORKFLOW_ENGINE=chat

This is an orchestration stub. #145 stays open for:

• message_history integration (DirectExecutor + adapter)
• Chat entrypoint + API routes
• load_recent_turns ordering fix
• Concurrent turn protection

Test plan

• 30 unit tests
• 7 integration tests
• All pre-existing tests pass

Partial implementation of #145. Stub lands the ABC wiring, turn management, and store integration.

🤖 Generated with Claude Code

[coderabbitai]

Warning

Review limit reached

Next included review available in 20 minutes.

View limit details

Limit details: You’ve used the included review currently available.

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

Learn how review limits work.

Review configuration:

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: e7dec58e-3820-4c71-b4ca-e6b14400fee9

📥 Commits

Reviewing files that changed from the base of the PR and between 43428fc and cac2e36.

📒 Files selected for processing (5)

• src/cloud_agents/workflow/executor/chat/__init__.py
• src/cloud_agents/workflow/executor/chat/runner.py
• src/cloud_agents/workflow/executor/factory.py
• tests/integration/test_chat_workflow_runner.py
• tests/unit/workflow/test_chat_workflow_runner.py

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[beesarmy]

Review: ChatWorkflowRunner — multi-turn chat as dynamic workflow

CI green, but two data corruption bugs are masked by in-memory test doubles that behave differently from the real PostgreSQL store.

Critical

[HIGH] Turn numbering breaks past max_context_turns — data overwrite

runner.py:159 — turn_number = len(prior_turns) counts loaded turns, not total turns. When a conversation exceeds max_context_turns (default 20), the next turn name collides with an existing one. The UPSERT silently overwrites the old turn's data.

Example: 25 turns exist (turn-0 to turn-24). load_recent_turns returns 20. len() = 20. Next turn named turn-20 — overwrites the real turn-20.

Fix: derive next turn number from the highest step_name in loaded turns (int(prior_turns[-1]["step_name"].split("-")[1]) + 1), or store a counter in run state.

[HIGH] load_recent_turns ordering mismatch — production vs tests

Real TranscriptStore.load_recent_turns SQL uses ORDER BY created_at DESC (newest first). In-memory test double returns chronological order (oldest first). get_history() would return messages in reverse order in production. All tests pass against the mock but would fail against real PostgreSQL.

Fix: add ORDER BY created_at ASC to the SQL (or reverse the result), since chronological is the natural expectation for conversation history.

Medium

[MED] No terminal state guard on send_message() — can send messages to cancelled/completed conversations. Check status before executing.

[MED] No concurrency protection — concurrent send_message() calls compute same turn number, UPSERT overwrites. Document or add asyncio lock per workflow_id.

[MED] Hardcoded timeout_seconds=600 — should be configurable via ChatWorkflowConfig.

[MED] send_message_stream() loses turns on exception — if run_stream() raises, _save_turn is never called. Wrap in try/finally.

Low (non-blocking)

• _save_turn skips assistant message for empty dict {} output (if result.output: should be is not None)
• Factory docstring doesn't mention ChatWorkflowRunner
• CHAT_MAX_CONTEXT_TURNS env var not validated for non-integer
• PR says "Closes #145" but doesn't implement ConversationStore ABC — clarify if approach changed

What looks good

• ✅ WorkflowRunner ABC correctly implemented (all abstract methods)
• ✅ Clean architecture — each turn is a step via StepExecutor
• ✅ send_message_stream() with real token-by-token streaming
• ✅ Context window management via max_context_turns
• ✅ 30 unit + 7 integration tests
• ✅ Factory integration (WORKFLOW_ENGINE=chat)
• ✅ OTEL tracing via StepMetadata

🤖 Reviewed by beesarmy via Claude Code

[jameswnl]

Review

The skeleton matches the later #145 shape (ChatWorkflowRunner as a WorkflowRunner, both stores, max_context_turns, send_message / send_message_stream). Tests cover the mocked happy path well.

The implementation does not yet do the thing #145 decided was the point of this runner: prior turns as structured message_history, with bounded, correctly ordered context, and a way to actually invoke it. Several of these will pass locally and fail against PostgreSQL.

Blockers

1. Prior turns are not conversation history — they are flattened DAG context

_build_context stuffs ConversationMessage lists into StepInput.context as prior-step outputs. DirectExecutor._build_messages / _build_user_prompt then stringify them under --- Prior step outputs --- and send a single new user message. pydantic-ai Agent.run(message_history=...) is never used.

That is exactly the gap the issue called out after the prototype:

Current StepInput.context flattens prior turns to text — this loses the structured role/content format LLMs need.

Tool calls/results, assistant role, and turn structure are lost. The LLM sees JSON blobs of previous outputs, not a chat. The adapter (ConversationMessage → ModelMessage) still does not exist, and DirectExecutor has no message_history path.

Until that lands, this is a turn logger around single-shot steps, not multi-turn chat.

2. Production load_recent_turns is newest-first; the runner and tests assume oldest-first

ORDER BY created_at DESC
LIMIT $2

(src/cloud_agents/storage/transcript_store.py)

The runner uses that list as-is for:

• turn_number = len(prior_turns) (ok until the window fills — see Podman socket path hardcodes macOS UID 501 — breaks Linux #3)
• _build_context insertion order → LLM sees newest turn first
• get_history() → messages come back reverse-chronological

The in-memory store in tests/integration/test_chat_workflow_runner.py returns entries[-limit:] (oldest-first). Unit tests stub chronological lists. Nothing asserts against the real SQL order. Against Postgres, get_history and context will be backwards.

Fix: reverse (or ORDER BY created_at ASC in a subquery) at the runner, and add a test that feeds DESC rows the way TranscriptStore actually returns them.

3. Turn names collide and upsert once max_context_turns is hit

turn_number = len(prior_turns)
turn_name = f"turn-{turn_number}"

After 20 turns with max_context_turns=20, every subsequent turn is named turn-20. TranscriptStore.save is ON CONFLICT (workflow_id, step_name) DO UPDATE, so you overwrite the same row forever and drop history.

Use list_steps() (or a dedicated counter / created_at max) for the next index, and keep load_recent_turns(limit=...) only for context. The in-memory test store appends on save, so this collision cannot show up in current tests.

4. WORKFLOW_ENGINE=chat is not wired to a process or API

• src/cloud_agents/entrypoint.py is still local vs else Temporal. WORKFLOW_ENGINE=chat boots the Temporal app.
• Local entrypoint always constructs LocalWorkflowRunner, not create_runner().
• Local/Temporal routers have no send_message / get_history routes.
• Chat start() ignores YAML definition; existing /run would be the wrong shape even if you swapped the runner.

Factory coverage is not enough. Either add a chat entrypoint + message routes, or do not advertise this as a WORKFLOW_ENGINE value yet (construct ChatWorkflowRunner from LCS / a dedicated router). Making chat mutually exclusive with local/temporal also fights the “chat is an option alongside workflows” note on #145.

5. send_message does not check that the conversation exists or is still open

get_status / cancel raise KeyError for missing IDs. send_message does not. A cancelled or unknown workflow_id still calls the LLM, then writes transcript rows with no matching run (or onto a cancelled conversation). CLI sessions already reject this (409 / RuntimeError). Check run state first (and refuse terminal statuses).

Should-fix before this is used for /query

Concurrent turns on the same conversation are undefined. #145 explicitly asked for last-write-wins vs 409. Two overlapping send_message calls will share a turn name and upsert. At least serialize per workflow_id (LocalWorkflowRunner already has _locks) or reject in-flight turns.

Identity / authz are dropped. start() writes authz_context={}. StepMetadata only sets conversation_id — not user_id / session_id even though start() stored them. Load them from the run on each turn.

result.transcript is discarded. _save_turn builds StepTranscript(events=[]). Tool-call audit that #145 wanted in the transcript view is thrown away.

Factory config is incomplete. ChatWorkflowConfig has tools, tools_module, mcp_servers; _create_chat never sets them. get_step_executor(step_def, self._spawner) also omits transcript_store, so spawn: ephemeral would not persist executor transcripts.

send_message and send_message_stream duplicate the load/build/dispatch sequence. Pull a _prepare_turn helper so turn naming and context cannot drift.

get_step_transcripts documents KeyError but never checks the run exists.

Tests / docs

• Tests labeled integration still mock get_step_executor. Fine as runner tests; they do not prove LLM context or Postgres ordering.
• Add cases for: missing/cancelled conversation, max_context_turns overflow naming, DESC load_recent_turns, concurrent sends, get_history order.
• ARCHITECTURE.md / DEPLOYMENT.md / entrypoint.py docstring still list two engines.
• Title uses RSPEED-0000 — replace with a real key before this goes near openshift/lightspeed-cloud-agents.

Suggested split

This PR is a reasonable orchestration stub. I would not merge it as closing #145 until:

1. Context is real message_history (adapter + DirectExecutor change, or a chat-specific path).
2. Turn indexing and load_recent_turns order are correct against TranscriptStore.
3. There is a concrete way to call this (API or documented constructor), including existence/terminal checks.

The factory chat engine and ABC wiring can land with those, or the runner can stay internal until the executor actually consumes history.

[beesarmy]

Re-Review: Fix commit cac2e36

3 of the critical/medium findings addressed. CI pending — approving based on fix quality.

Finding	Fix
[HIGH] Turn numbering breaks past max_context_turns	✅ _next_turn_number() parses highest turn-N from loaded turns instead of len(). Prevents collision.
[HIGH] load_recent_turns ordering mismatch	⚠️ Not addressed in this commit — track as follow-up
[MED] No terminal state guard	✅ _check_not_terminal() called at top of both send_message() and send_message_stream(). Raises RuntimeError for cancelled/completed.
[MED] Hardcoded timeout	✅ timeout_seconds added to ChatWorkflowConfig, wired through _build_step_input()
[MED] Concurrency	⚠️ Not addressed — track as follow-up
[MED] Stream exception loses turns	⚠️ Not addressed — track as follow-up

The _next_turn_number() implementation is clean — handles edge cases (no turns, malformed names, non-sequential numbering). The terminal state guard is correctly applied to both blocking and streaming paths.

Approved. The 3 remaining items (ordering, concurrency, stream exception) are real but deferrable.

🤖 Reviewed by beesarmy via Claude Code

[jameswnl]

Acknowledged — merging as orchestration stub, gaps tracked

Items 3 (turn naming) and 5 (terminal guard) are already fixed in cac2e36.

Remaining gaps tracked as follow-ups (do NOT close #145 on merge):

1. message_history integration — DirectExecutor needs a message_history path via pydantic-ai Agent.run(message_history=...). The ConversationMessage → ModelMessage adapter is needed. This is a DirectExecutor change, not a ChatWorkflowRunner change.
2. load_recent_turns ordering — fix SQL to return ASC (or reverse in runner). Add a test with DESC input.
3. Chat entrypoint + API routes — /chat/{conv_id}/message route, or document constructor-only usage for embedded mode.
4. Concurrent turn protection — per-workflow asyncio lock (same pattern as LocalWorkflowRunner).
5. Identity propagation — load user_id/session_id from RunStateStore on each turn, set on StepMetadata.
6. result.transcript not discarded — pass exec_result.transcript to StepTranscript.events.

This PR is a valid orchestration stub. #145 stays open until message_history is wired.