Task-Driven Workflow

Tasks aren't just todos — they're the backbone of persistent LLM workflows. By creating tasks for multi-step work, you ensure continuity across sessions and provide audit trails for what was done.

Why Task-Driven?

Without tasks:

• LLM starts each session unsure what to do
• Multi-step work is forgotten mid-execution
• No record of what's been done

With tasks:

• LLM resumes in-progress tasks immediately
• Multi-step work survives across sessions
• Built-in audit trail of all work

The Pattern

1. At session start: check in_progress tasks
2. If tasks exist: resume them
3. If no tasks: create new tasks for current work
4. Update task status as you progress
5. Mark done when complete

Implementation

Step 1: Create a task for multi-step work

def start_workflow(title, steps):
    """Create a task for multi-step work."""
    task_id = create_task(
        title=title,
        description=f"Steps:\n" + "\n".join(f"  {i+1}. {s}" for i, s in enumerate(steps)),
        priority="high"
    )
    return task_id

# Example
task_id = start_workflow("Deploy Synapse v1.6.0", [
    "Bump version in package.json",
    "Update CHANGELOG.md",
    "Commit and push",
    "Wait for CI green",
    "Verify deployment"
])

Step 2: Track progress in task description

def update_progress(task_id, current_step, total_steps, status_note):
    """Update task with current progress."""
    description = f"Progress: {current_step}/{total_steps}\nStatus: {status_note}"
    update_task(task_id, status="in_progress", description=description)

# Example
update_progress(task_id, 2, 5, "CHANGELOG updated, committing now")

Step 3: Resume across sessions

def resume_work():
    """At session start, find and resume in-progress tasks."""
    tasks = list_tasks(status="in_progress")
    
    for task in tasks:
        print(f"Resuming: {task['title']}")
        print(f"Last status: {task['description']}")
        
        # Parse progress from description
        progress = parse_progress(task['description'])
        next_step = progress['current_step'] + 1
        
        # Continue from next step
        continue_from_step(task['id'], next_step)

Step 4: Complete and archive

def complete_task(task_id, summary):
    """Mark task done with completion summary."""
    update_task(task_id, 
        status="done",
        description=f"COMPLETED. Summary: {summary}"
    )
    # Also store as memory for long-term reference
    remember(
        category="project",
        key=f"completed_{task_id}",
        content=f"Task: {task_id}\nSummary: {summary}",
        tags=["completed", "task"],
        priority="normal"
    )

Full Example: Deploy Workflow

class DeployWorkflow:
    def __init__(self, version):
        self.version = version
        self.task_id = None
        self.steps = [
            ("Bump version", self.bump_version),
            ("Update changelog", self.update_changelog),
            ("Commit and push", self.commit_push),
            ("Wait for CI", self.wait_for_ci),
            ("Verify deployment", self.verify_deployment),
        ]
    
    def run(self):
        # Check if already in progress
        existing = self.find_existing()
        if existing:
            self.task_id = existing['id']
            start_step = self.parse_progress(existing['description'])
        else:
            self.task_id = create_task(
                title=f"Deploy Synapse v{self.version}",
                description=self.build_description(0),
                priority="high"
            )
            start_step = 0
        
        # Execute remaining steps
        for i in range(start_step, len(self.steps)):
            step_name, step_fn = self.steps[i]
            self.update_progress(i, f"Running: {step_name}")
            try:
                step_fn()
            except Exception as e:
                self.update_progress(i, f"FAILED at {step_name}: {e}")
                raise
        
        self.complete()
    
    def update_progress(self, step_idx, status):
        update_task(self.task_id,
            status="in_progress",
            description=f"Step {step_idx+1}/{len(self.steps)}: {status}"
        )
    
    def complete(self):
        complete_task(self.task_id, f"Deployed v{self.version} successfully")

Task Hierarchy

For complex work, use parent-child task relationships:

# Parent task
parent_id = create_task("v1.6.0 Release", priority="high")

# Sub-tasks (linked via tags)
create_task("Bump version", 
    description=f"Parent: {parent_id}",
    tags=["v1.6.0", f"parent-{parent_id}"],
    priority="high")

create_task("Update docs",
    description=f"Parent: {parent_id}",
    tags=["v1.6.0", f"parent-{parent_id}"],
    priority="normal")

Search for sub-tasks:

curl -H "Authorization: Bearer $KEY" \
     ".../memory/search?q=parent-{parent_id}&tag=v1.6.0"

Status Workflow

pending → in_progress → done
                ↘ cancelled

Pending

Task created but not started. Use for planned work.

In Progress

Currently being worked on. Update description with progress.

Done

Completed successfully. Description should include summary.

Cancelled

Abandoned. Description should include reason.

Best Practices

Common Patterns

Pattern: Bug Fix Workflow

def fix_bug(bug_id, description):
    task_id = create_task(
        title=f"Fix bug {bug_id}",
        description=description,
        priority="high"
    )
    
    # Investigate
    update_progress(task_id, "Investigating root cause")
    root_cause = investigate()
    
    # Fix
    update_progress(task_id, f"Applying fix: {root_cause}")
    apply_fix(root_cause)
    
    # Test
    update_progress(task_id, "Testing fix")
    run_tests()
    
    # Deploy
    update_progress(task_id, "Deploying fix")
    deploy()
    
    complete_task(task_id, f"Fixed: {root_cause}")

Pattern: Research Workflow

def research_topic(topic):
    task_id = create_task(
        title=f"Research: {topic}",
        priority="normal"
    )
    
    update_progress(task_id, "Gathering sources")
    sources = gather_sources(topic)
    
    update_progress(task_id, "Analyzing")
    analysis = analyze(sources)
    
    update_progress(task_id, "Storing findings")
    remember("fact", f"research_{topic}", analysis,
             tags=["research", topic], priority="normal")
    
    complete_task(task_id, f"Research complete: {len(sources)} sources")

Next Steps

• Session Start Pattern
• Chat Polling Pattern
• Error Recovery