refactor: consolidate triplicated CSV/text loop-task expansion in process.py#1957
refactor: consolidate triplicated CSV/text loop-task expansion in process.py#1957praisonai-triage-agent[bot] wants to merge 2 commits into
Conversation
…cess.py This consolidates three near-identical implementations of CSV/text-file loop-task expansion onto the existing _create_loop_subtasks() helper: - Enhanced _create_loop_subtasks() with optional decision_mode parameter - Replaced inline start-task logic (lines 1116-1205) with helper call - Replaced inline mid-workflow logic (lines 1273-1330) with helper call Removes ~300 lines of duplicated logic and fixes latent CSV-parsing inconsistency where inline copies used plain csv.reader() instead of csv.reader(f, quotechar='"', escapechar='\\'). All loop feature behavior preserved: CSV/text files, decision-task creation with inherited next_tasks, and sync/async paths work identically. Fixes #1953 Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>
|
@coderabbitai review |
|
/review |
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
✅ Action performedReview finished.
|
|
Important Review skippedBot user detected. To trigger a single review, invoke the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughThis PR consolidates three duplicate loop-task expansion implementations in ChangesLoop Subtask Consolidation
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
|
@copilot Do a thorough review of this PR. Read ALL existing reviewer comments above from Qodo, Coderabbit, and Gemini first — incorporate their findings. Review areas:
|
Greptile SummaryThis PR consolidates three near-identical CSV/text-file loop-task expansion code blocks into the existing
Confidence Score: 3/5The routing logic for edge-case workflows is fragile and the accumulated issues can corrupt task routing before merge. The consolidation removes ~300 lines of duplicated code cleanly, but the start-task search still relies on a name.startswith() filter that can pull in unrelated sibling tasks, is_start is computed from the raw line index rather than a non-blank counter in the text-file branch so the flag is wrong whenever the file has leading blank lines, and inherited_next_tasks is aliased into both next_tasks and condition["done"] on every CSV decision subtask leaving both fields vulnerable to a single in-place mutation. src/praisonai-agents/praisonaiagents/process/process.py — specifically the _create_loop_subtasks() helper (text-file branch task indexing, list aliasing) and both call sites (name-prefix subtask search). Important Files Changed
Sequence DiagramsequenceDiagram
participant W as Workflow Engine
participant H as _create_loop_subtasks()
participant T as self.tasks dict
Note over W: Start-task path (decision_mode=True)
W->>H: "_create_loop_subtasks(loop_task, decision_mode=True)"
H->>T: "Add subtasks type=decision with condition done and retry"
H->>T: Chain subtasks via next_tasks and condition done
H-->>W: returns
W->>T: Filter tasks by name prefix
W->>W: "start_task set to first subtask with is_start=True"
W->>W: "parent_loop_task.status = completed"
Note over W: Runtime path (decision_mode=False)
W->>H: "_create_loop_subtasks(current_task, decision_mode=False)"
H->>T: "Add subtasks type=task"
H->>T: Chain subtasks via next_tasks
H-->>W: returns
W->>T: Filter tasks by name prefix
W->>W: current_task.next_tasks set to first subtask name
W->>W: "current_task._subtasks_created = True"
Note over W: Decision subtask routing CSV decision_mode=True
W->>W: subtask executes result becomes decision_str
W->>W: condition done routes to next subtask or parent next_tasks
W->>W: condition retry routes back to self via row_name
W->>W: condition exit with empty list exits workflow
Reviews (2): Last reviewed commit: "fix: resolve critical issues in loop tas..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/praisonai-agents/praisonaiagents/process/process.py`:
- Around line 1175-1182: Replace lexicographic sorting of generated subtasks
with selection by creation/order or the explicit is_start flag: in
src/praisonai-agents/praisonaiagents/process/process.py (lines 1175-1182) stop
doing subtasks.sort(key=lambda t: t.name) and instead pick the subtask that has
is_start == True (or the earliest-created subtask preserved by insertion order
of self.tasks) to set start_task; likewise in
src/praisonai-agents/praisonaiagents/process/process.py (lines 1249-1256) change
the logic that wires current_task.next_tasks to the “first” runtime-generated
subtask to use the same non-lexicographic selection (is_start or creation order)
rather than sorting by t.name so tasks like _2 vs _10 are handled correctly.
- Around line 1171-1183: The parent loop task isn't retired after expansion
causing it to remain "not started" and block completion; in the block that calls
_create_loop_subtasks(start_task, decision_mode=True) (and then finds subtasks),
capture the original parent task identifier (e.g., parent_name =
start_task.name), then remove or mark that parent task as completed/retired in
self.tasks (e.g., del self.tasks[parent_name] or set its status to a terminal
state) immediately after you switch start_task to the first generated subtask;
this ensures _check_all_tasks_completed_sync() can succeed and
_find_next_not_started_task() won't re-expand the same loop. Ensure any logging
that references start_task.input_file uses the correct variable (use the
captured parent input_file if you log the original parent) so you don't
accidentally log the new subtask's input_file.
- Around line 250-253: The decision-mode condition dict is using placeholder
names ("current"/"next") that the router doesn't resolve; update the dict
construction (the condition passed where "done"/"retry"/"exit" are set) to use
concrete task names: set "retry" to a list containing the actual current task
identifier (e.g., current_task_name or current_task.id/name used in this scope)
instead of ["current"], and set the fallback for "done" to inherited_next_tasks
if present otherwise to the real next task name list (e.g., next_tasks) instead
of ["next"]; keep "exit": [] as-is. Ensure you reference the existing variables
inherited_next_tasks, next_tasks and whichever symbol represents the current
task (current_task_name/current_task.id/current_task.name) in that code block.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 63c4310d-c38d-4eb3-bb12-97633ce83774
📒 Files selected for processing (1)
src/praisonai-agents/praisonaiagents/process/process.py
| condition={ | ||
| "done": inherited_next_tasks if inherited_next_tasks else ["next"], # Use full inherited_next_tasks | ||
| "retry": ["current"], | ||
| "exit": [] # Empty list for exit condition |
There was a problem hiding this comment.
Use concrete task names in decision-mode conditions.
At Line 1451 the router only resolves actual task names (or exit). The new "retry": ["current"] / fallback "done": ["next"] placeholders are never interpreted, so a retry on these CSV decision subtasks falls through to next_tasks and advances to the next row instead of rerunning the current one.
Proposed fix
- inherited_next_tasks = loop_task.next_tasks if loop_task.next_tasks else []
+ inherited_next_tasks = loop_task.next_tasks if loop_task.next_tasks else []
+ row_name = f"{loop_task.name}_{task_count}" if loop_task.name else task_desc
row_task = Task(
description=f"{loop_task.description}\n{task_desc}" if loop_task.description else task_desc,
agent=loop_task.agent,
- name=f"{loop_task.name}_{task_count}" if loop_task.name else task_desc,
+ name=row_name,
expected_output=getattr(loop_task, 'expected_output', None),
on_task_complete=loop_task.callback, # Inherit callback from parent loop task
is_start=(task_count == 1),
task_type="decision", # Change to decision type for start tasks
next_tasks=inherited_next_tasks, # Inherit parent's next tasks
condition={
- "done": inherited_next_tasks if inherited_next_tasks else ["next"], # Use full inherited_next_tasks
- "retry": ["current"],
+ "done": inherited_next_tasks if inherited_next_tasks else [],
+ "retry": [row_name],
"exit": [] # Empty list for exit condition
}
)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-agents/praisonaiagents/process/process.py` around lines 250 -
253, The decision-mode condition dict is using placeholder names
("current"/"next") that the router doesn't resolve; update the dict construction
(the condition passed where "done"/"retry"/"exit" are set) to use concrete task
names: set "retry" to a list containing the actual current task identifier
(e.g., current_task_name or current_task.id/name used in this scope) instead of
["current"], and set the fallback for "done" to inherited_next_tasks if present
otherwise to the real next task name list (e.g., next_tasks) instead of
["next"]; keep "exit": [] as-is. Ensure you reference the existing variables
inherited_next_tasks, next_tasks and whichever symbol represents the current
task (current_task_name/current_task.id/current_task.name) in that code block.
| if start_task and start_task.task_type == "loop" and getattr(start_task, "input_file", None): | ||
| try: | ||
| file_ext = os.path.splitext(start_task.input_file)[1].lower() | ||
| new_tasks = [] | ||
|
|
||
| if file_ext == ".csv": | ||
| with open(start_task.input_file, "r", encoding="utf-8") as f: | ||
| reader = csv.reader(f, quotechar='"', escapechar='\\') # Handle quoted/escaped fields | ||
| previous_task = None | ||
| task_count = 0 | ||
|
|
||
| for i, row in enumerate(reader): | ||
| if not row: # Skip truly empty rows | ||
| continue | ||
|
|
||
| # Properly handle Q&A pairs with potential commas | ||
| task_desc = row[0].strip() if row else "" | ||
| if len(row) > 1: | ||
| # Preserve all fields in case of multiple commas | ||
| question = row[0].strip() | ||
| answer = ",".join(field.strip() for field in row[1:]) | ||
| task_desc = f"Question: {question}\nAnswer: {answer}" | ||
|
|
||
| if not task_desc: # Skip rows with empty content | ||
| continue | ||
|
|
||
| task_count += 1 | ||
| logging.debug(f"Processing CSV row {i+1}: {task_desc}") | ||
|
|
||
| # Inherit next_tasks from parent loop task | ||
| inherited_next_tasks = start_task.next_tasks if start_task.next_tasks else [] | ||
|
|
||
| row_task = Task( | ||
| description=f"{start_task.description}\n{task_desc}" if start_task.description else task_desc, | ||
| agent=start_task.agent, | ||
| name=f"{start_task.name}_{task_count}" if start_task.name else task_desc, | ||
| expected_output=getattr(start_task, 'expected_output', None), | ||
| on_task_complete=start_task.callback, # Inherit callback from parent loop task | ||
| is_start=(task_count == 1), | ||
| task_type="decision", # Change to decision type | ||
| next_tasks=inherited_next_tasks, # Inherit parent's next tasks | ||
| condition={ | ||
| "done": inherited_next_tasks if inherited_next_tasks else ["next"], # Use full inherited_next_tasks | ||
| "retry": ["current"], | ||
| "exit": [] # Empty list for exit condition | ||
| } | ||
| ) | ||
| self.tasks[row_task.id] = row_task | ||
| new_tasks.append(row_task) | ||
|
|
||
| if previous_task: | ||
| previous_task.next_tasks = [row_task.name] | ||
| previous_task.condition["done"] = [row_task.name] # Use "done" consistently | ||
| previous_task = row_task | ||
|
|
||
| # For the last task in the loop, ensure it points to parent's next tasks | ||
| if task_count > 0 and not row_task.next_tasks: | ||
| row_task.next_tasks = inherited_next_tasks | ||
|
|
||
| logging.info(f"Processed {task_count} rows from CSV file") | ||
| else: | ||
| # If not CSV, read lines | ||
| with open(start_task.input_file, "r", encoding="utf-8") as f: | ||
| lines = f.read().splitlines() | ||
| previous_task = None | ||
| for i, line in enumerate(lines): | ||
| row_task = Task( | ||
| description=f"{start_task.description}\n{line.strip()}" if start_task.description else line.strip(), | ||
| agent=start_task.agent, | ||
| name=f"{start_task.name}_{i+1}" if start_task.name else line.strip(), | ||
| expected_output=getattr(start_task, 'expected_output', None), | ||
| on_task_complete=start_task.callback, # Inherit callback from parent loop task | ||
| is_start=(i == 0), | ||
| task_type="task", | ||
| condition={ | ||
| "complete": ["next"], | ||
| "retry": ["current"] | ||
| } | ||
| ) | ||
| self.tasks[row_task.id] = row_task | ||
| new_tasks.append(row_task) | ||
|
|
||
| if previous_task: | ||
| previous_task.next_tasks = [row_task.name] | ||
| previous_task.condition["complete"] = [row_task.name] | ||
| previous_task = row_task | ||
|
|
||
| if new_tasks: | ||
| start_task = new_tasks[0] | ||
| logging.info(f"Created {len(new_tasks)} tasks from: {start_task.input_file}") | ||
| self._create_loop_subtasks(start_task, decision_mode=True) | ||
| # Get the first created subtask as the new start task | ||
| subtasks = [ | ||
| t for t in self.tasks.values() | ||
| if t.name.startswith(start_task.name + "_") | ||
| ] | ||
| if subtasks: | ||
| # Sort by name to get the first subtask | ||
| subtasks.sort(key=lambda t: t.name) | ||
| start_task = subtasks[0] | ||
| logging.info(f"Created {len(subtasks)} tasks from: {start_task.input_file}") |
There was a problem hiding this comment.
Retire the original start loop task after expansion.
This block switches start_task to the first generated subtask, but the parent loop task stays in self.tasks as "not started". That means _check_all_tasks_completed_sync() can never succeed while it is pending, and _find_next_not_started_task() can rediscover it later and expand the same file a second time.
Proposed fix
if start_task and start_task.task_type == "loop" and getattr(start_task, "input_file", None):
try:
- self._create_loop_subtasks(start_task, decision_mode=True)
+ parent_loop_task = start_task
+ self._create_loop_subtasks(parent_loop_task, decision_mode=True)
# Get the first created subtask as the new start task
subtasks = [
t for t in self.tasks.values()
- if t.name.startswith(start_task.name + "_")
+ if t.name.startswith(parent_loop_task.name + "_")
]
if subtasks:
- # Sort by name to get the first subtask
- subtasks.sort(key=lambda t: t.name)
- start_task = subtasks[0]
- logging.info(f"Created {len(subtasks)} tasks from: {start_task.input_file}")
+ parent_loop_task._subtasks_created = True
+ parent_loop_task.status = "completed"
+ start_task = next((t for t in subtasks if t.is_start), subtasks[0])
+ logging.info(f"Created {len(subtasks)} tasks from: {parent_loop_task.input_file}")
except Exception as e:
logging.error(f"Failed to read file tasks: {e}")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-agents/praisonaiagents/process/process.py` around lines 1171 -
1183, The parent loop task isn't retired after expansion causing it to remain
"not started" and block completion; in the block that calls
_create_loop_subtasks(start_task, decision_mode=True) (and then finds subtasks),
capture the original parent task identifier (e.g., parent_name =
start_task.name), then remove or mark that parent task as completed/retired in
self.tasks (e.g., del self.tasks[parent_name] or set its status to a terminal
state) immediately after you switch start_task to the first generated subtask;
this ensures _check_all_tasks_completed_sync() can succeed and
_find_next_not_started_task() won't re-expand the same loop. Ensure any logging
that references start_task.input_file uses the correct variable (use the
captured parent input_file if you log the original parent) so you don't
accidentally log the new subtask's input_file.
| subtasks = [ | ||
| t for t in self.tasks.values() | ||
| if t.name.startswith(start_task.name + "_") | ||
| ] | ||
| if subtasks: | ||
| # Sort by name to get the first subtask | ||
| subtasks.sort(key=lambda t: t.name) | ||
| start_task = subtasks[0] |
There was a problem hiding this comment.
Don't derive loop order from lexicographic task names. Sorting generated subtask names will pick _10 before _2, so files with 10+ rows/lines can start from the wrong subtask in both startup and runtime expansion paths.
src/praisonai-agents/praisonaiagents/process/process.py#L1175-L1182: select the generated subtask withis_start(or preserve helper creation order) instead ofsubtasks.sort(key=lambda t: t.name).src/praisonai-agents/praisonaiagents/process/process.py#L1249-L1256: use the same non-lexicographic selection when wiringcurrent_task.next_tasksto the first runtime-generated subtask.
📍 Affects 1 file
src/praisonai-agents/praisonaiagents/process/process.py#L1175-L1182(this comment)src/praisonai-agents/praisonaiagents/process/process.py#L1249-L1256
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-agents/praisonaiagents/process/process.py` around lines 1175 -
1182, Replace lexicographic sorting of generated subtasks with selection by
creation/order or the explicit is_start flag: in
src/praisonai-agents/praisonaiagents/process/process.py (lines 1175-1182) stop
doing subtasks.sort(key=lambda t: t.name) and instead pick the subtask that has
is_start == True (or the earliest-created subtask preserved by insertion order
of self.tasks) to set start_task; likewise in
src/praisonai-agents/praisonaiagents/process/process.py (lines 1249-1256) change
the logic that wires current_task.next_tasks to the “first” runtime-generated
subtask to use the same non-lexicographic selection (is_start or creation order)
rather than sorting by t.name so tasks like _2 vs _10 are handled correctly.
|
@claude You are the FINAL architecture reviewer. If the branch is under MervinPraison/PraisonAI (not a fork), you are able to make modifications to this branch and push directly. SCOPE: Focus ONLY on Python packages (praisonaiagents, praisonai). Do NOT modify praisonai-rust or praisonai-ts. Read ALL comments above from Gemini, Qodo, CodeRabbit, and Copilot carefully before responding. Phase 1: Review per AGENTS.md
Phase 2: FIX Valid Issues Phase 3: Final Verdict |
|
Claude finished @MervinPraison's task —— View job Final Architecture Review - PR #1957 ✅Phase 1: Architecture Review per AGENTS.md
Phase 2: Fix Valid Issues from Reviewers
Phase 3: Final Verdict
✅ APPROVED - READY TO MERGEArchitecture Analysis✅ Protocol-driven design: The consolidation correctly moves shared logic into a helper method while keeping the core workflow logic intact. This follows the AGENTS.md guideline of avoiding code duplication. ✅ Backward compatibility: The changes preserve all existing loop behavior through the ✅ Performance: No hot-path regressions detected - the consolidation reduces code complexity and maintains the same execution flow. Issues Fixed🔧 FIXED - Critical Issues:
📝 Skipped - Medium Priority Issues:
Testing✅ Syntax validation: All Python syntax checks pass Final VerdictThe PR is architecturally sound and all critical issues have been resolved. The consolidation successfully:
Recommendation: MERGE - This PR significantly improves code maintainability while preserving all functionality. Fixes pushed to branch: claude/issue-1953-20260613-1158 |
- Fix lexicographic sorting that breaks ordering with 10+ subtasks - Retire parent loop task after expansion to prevent completion blocking - Replace placeholder task names in decision conditions with concrete names - Use is_start flag instead of name sorting for proper subtask selection Addresses reviewer feedback from Greptile and CodeRabbit on PR #1957 Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>
| if decision_mode: | ||
| # For start tasks: create decision-type tasks with conditions and inherited next_tasks | ||
| inherited_next_tasks = loop_task.next_tasks if loop_task.next_tasks else [] | ||
| row_name = f"{loop_task.name}_{task_count}" if loop_task.name else task_desc | ||
| row_task = Task( | ||
| description=f"{loop_task.description}\n{task_desc}" if loop_task.description else task_desc, | ||
| agent=loop_task.agent, | ||
| name=row_name, | ||
| expected_output=getattr(loop_task, 'expected_output', None), | ||
| on_task_complete=loop_task.callback, # Inherit callback from parent loop task | ||
| is_start=(task_count == 1), | ||
| task_type="decision", # Change to decision type for start tasks | ||
| next_tasks=inherited_next_tasks, # Inherit parent's next tasks | ||
| condition={ | ||
| "done": inherited_next_tasks if inherited_next_tasks else [], | ||
| "retry": [row_name], | ||
| "exit": [] # Empty list for exit condition | ||
| } | ||
| ) |
There was a problem hiding this comment.
inherited_next_tasks aliased into both next_tasks and condition["done"] on the same task
inherited_next_tasks is assigned loop_task.next_tasks by reference (line 240). It is then passed as both next_tasks=inherited_next_tasks and as the value of condition["done"] in the constructor. All three variables — loop_task.next_tasks, the subtask's next_tasks field, and condition["done"] — now point to the same list object.
Any in-place mutation of one of these (e.g. task.next_tasks.append(x) or task.next_tasks[0] = y) would silently corrupt all three simultaneously. The chaining loop at line 273 does use reassignment (previous_task.next_tasks = [row_task.name]), which is safe, but callers outside this helper that receive the subtask object could mutate the list. A defensive copy would eliminate the risk: inherited_next_tasks = list(loop_task.next_tasks) if loop_task.next_tasks else [].
1 participant
Consolidates three near-identical implementations of CSV/text-file loop-task expansion onto the existing _create_loop_subtasks() helper, removing ~300 lines of duplicated logic and fixing a latent CSV-parsing inconsistency. Enhanced _create_loop_subtasks() with optional decision_mode parameter. All loop feature behavior preserved. Fixes #1953
Summary by CodeRabbit