feat: Add conversation variable persistence layer

[tomerqodo]

Benchmark PR from agentic-review-benchmarks#4

[greptile-apps]

Greptile Overview

Greptile Summary

This PR refactors conversation variable persistence by extracting database operations from workflow nodes into a dedicated persistence layer, following clean architecture principles. The variable assigner nodes (v1 and v2) were updated to remove inline database calls, delegating persistence to ConversationVariablePersistenceLayer which listens to node execution events.

Major Changes:

• Created new ConversationVariablePersistenceLayer that intercepts NodeRunSucceededEvent from variable assigner nodes
• Moved database operations from core/workflow/nodes/variable_assigner/common/impl.py to services/conversation_variable_updater.py
• Updated AdvancedChatAppRunner to initialize conversation variables and attach the persistence layer
• Removed database dependencies from variable assigner nodes, improving separation of concerns
• Updated import linter rules to reflect new architecture

Critical Issues:

• ConversationVariableUpdaterImpl creates database sessions without using context managers, violating rule feat: allow pass hostname in docker env #6 and causing resource leaks
• The persistence layer calls flush() immediately after each update(), defeating the batching design of the updater protocol

Confidence Score: 1/5

• This PR contains a critical database session leak that will cause production issues
• The session management bug in ConversationVariableUpdaterImpl is a blocker that violates coding standards and will cause database connection leaks in production. Every conversation variable update creates an unclosed session, which will exhaust the connection pool under load.
• Critical attention needed for api/services/conversation_variable_updater.py - must fix session context manager before merge

Important Files Changed

Filename	Overview
api/services/conversation_variable_updater.py	New file with critical session management bug - Session not used with context manager
api/core/app/layers/conversation_variable_persist_layer.py	New persistence layer that immediately flushes after each update, defeating batching
api/core/workflow/nodes/variable_assigner/v1/node.py	Removed inline database operations, moved to persistence layer
api/core/workflow/nodes/variable_assigner/v2/node.py	Removed inline database operations, moved to persistence layer
api/core/app/apps/advanced_chat/app_runner.py	Added conversation variable persistence layer and initialization logic

Sequence Diagram

sequenceDiagram
    participant User
    participant AppRunner as AdvancedChatAppRunner
    participant Layer as ConversationVariablePersistenceLayer
    participant Node as VariableAssignerNode
    participant Updater as ConversationVariableUpdaterImpl
    participant DB as Database

    User->>AppRunner: Run workflow with conversation
    AppRunner->>AppRunner: Initialize conversation variables from DB
    AppRunner->>Layer: Create persistence layer with updater
    AppRunner->>AppRunner: Attach layer to workflow engine
    
    AppRunner->>Node: Execute variable assigner node
    Node->>Node: Update variable in variable pool
    Node->>Node: Add updated variable to process_data
    Node-->>Layer: Emit NodeRunSucceededEvent
    
    Layer->>Layer: Check if event is VariableAssigner
    Layer->>Layer: Extract updated variables from process_data
    Layer->>Layer: Filter for conversation variables only
    
    loop For each conversation variable
        Layer->>Updater: update(conversation_id, variable)
        Updater->>DB: Create Session (without context manager!)
        Updater->>DB: SELECT ConversationVariable
        Updater->>DB: UPDATE data field
        Updater->>DB: COMMIT (session not closed!)
        Layer->>Updater: flush()
        Note over Updater: flush() is a no-op
    end
    
    Note over Updater,DB: Session leak: Session never closed!

Loading

[greptile-apps]

_{2 files reviewed, 2 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Session not used as context manager, causing resource leak

Suggested change

	session = Session(db.engine)
	row = session.scalar(stmt)
	if not row:
	raise ConversationVariableNotFoundError("conversation variable not found in the database")
	row.data = variable.model_dump_json()
	session.commit()
	with Session(db.engine, expire_on_commit=False) as session:
	row = session.scalar(stmt)
	if not row:
	raise ConversationVariableNotFoundError("conversation variable not found in the database")
	row.data = variable.model_dump_json()
	session.commit()

Prompt To Fix With AI

This is a comment left during a code review.
Path: api/services/conversation_variable_updater.py
Line: 18:23

Comment:
Session not used as context manager, causing resource leak

```suggestion
        with Session(db.engine, expire_on_commit=False) as session:
            row = session.scalar(stmt)
            if not row:
                raise ConversationVariableNotFoundError("conversation variable not found in the database")
            row.data = variable.model_dump_json()
            session.commit()
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

flush() called after each update() defeats batching purpose. The ConversationVariableUpdater protocol allows batching, but calling flush immediately after each update negates any batching benefits and creates unnecessary database commits in a loop.

Prompt To Fix With AI

This is a comment left during a code review.
Path: api/core/app/layers/conversation_variable_persist_layer.py
Line: 52:53

Comment:
`flush()` called after each `update()` defeats batching purpose. The `ConversationVariableUpdater` protocol allows batching, but calling flush immediately after each update negates any batching benefits and creates unnecessary database commits in a loop.

How can I resolve this? If you propose a fix, please make it concise.