#4496 Agentic Task Queues with Folder-Based Markdown Files; Event-Driven (w/.listen() method)

I want to develop an agentic framework where agents are assigned tasks and save their outputs to specified folders as markdown files. ... ... ... ...
I would expect follow-on agents to search the specified folder for available (i.e. "waiting") markdown folders, and get their assignments in that method.
Think innovatively. How might blockchain fit into this framework?

#4496 I want to develop an agentic framework where agents are assigned tasks and save their outputs to specified folders as markdown files. Other agents can "pick up" the markdown file and do additional work then save the update and move the markdown file to the next specified folder. Basically: agents do work and save their work in a folder, then other agents pick up where they left off and stage the results "forward"

What you are describing is a sequential, file-based pipeline (or blackboard architecture) for multi-agent systems. It’s a brilliant, highly decoupled way to build agents because it gives you a built-in audit trail. If an agent hallucinates or fails, you can look exactly inside the folder where it broke and see the exact state of the Markdown file.

Here is a conceptual blueprint and a lightweight Python implementation using standard libraries (plus Frontmatter for agent metadata) to get you started.

🏛️ The Architecture: "The Assembly Line"

To make this work seamlessly, your Markdown files should use YAML Frontmatter. This allows agents to read and write metadata (like who processed it last, the status, or the next required step) without messing up the actual content of the report.

📁 The Folder Structure

agent_pipeline/
│
├── 01_backlog/        # Raw prompts, ideas, or initial requests
├── 02_researched/     # Researcher agent drops files here
├── 03_drafted/        # Writer agent picks up from 02, saves here
├── 04_reviewed/       # Editor agent reviews, saves final copy here
└── archive/           # Completed workflows

🛠️ Code Implementation

Here is a robust foundation for orchestrating this workflow. It defines a base Agent class that handles the file I/O, moving files "forward" in the pipeline automatically.

import os
import shutil
from pathlib import Path
import yaml  # pip install pyyaml

class FilePipelineAgent:
    def __init__(self, name: str, input_dir: str, output_dir: str):
        self.name = name
        self.input_dir = Path(input_dir)
        self.output_dir = Path(output_dir)
        
        # Ensure directories exist
        self.input_dir.mkdir(parents=True, exist_ok=True)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def parse_markdown(self, file_path: Path) -> tuple[dict, str]:
        """Splits a markdown file into its YAML frontmatter and content."""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        if content.startswith('---'):
            parts = content.split('---', 2)
            if len(parts) >= 3:
                metadata = yaml.safe_load(parts[1]) or {}
                body = parts[2].strip()
                return metadata, body
        return {}, content.strip()

    def save_markdown(self, filename: str, metadata: dict, body: str):
        """Saves updated data into the agent's output directory."""
        target_path = self.output_dir / filename
        
        # Update metadata history
        if "history" not in metadata:
            metadata["history"] = []
        metadata["history"].append(f"Processed by {self.name}")
        metadata["current_stage"] = self.output_dir.name

        frontmatter = yaml.dump(metadata, default_flow_style=False)
        
        with open(target_path, 'w', encoding='utf-8') as f:
            f.write(f"---\n{frontmatter}---\n\n{body}")
        
        print(f"🤖 [{self.name}] Saved updated work to: {target_path}")

    def process_next_file(self):
        """Looks for the first available markdown file in the input directory."""
        files = list(self.input_dir.glob("*.md"))
        if not files:
            print(f"💤 [{self.name}] No files to process in {self.input_dir.name}/")
            return False
        
        # Pick up the first file
        file_to_process = files[0]
        print(f"🎯 [{self.name}] Found file: {file_to_process.name}")
        
        # 1. Read the file
        metadata, body = self.parse_markdown(file_to_process)
        
        # 2. DO THE WORK (To be overridden by specific LLM logic)
        updated_metadata, updated_body = self.execute_llm_task(metadata, body)
        
        # 3. Save to the NEXT folder
        self.save_markdown(file_to_process.name, updated_metadata, updated_body)
        
        # 4. Remove from the OLD folder (Staging it forward)
        file_to_process.unlink()
        return True

    def execute_llm_task(self, metadata: dict, body: str) -> tuple[dict, str]:
        """Placeholder: Override this method for your specific agents."""
        raise NotImplementedError("Subclasses must implement their own LLM logic.")

🤖 Example: Creating Specific Agents

Now we can inherit from our base pipeline class and inject our LLM calls (using OpenAI, Anthropic, Ollama, etc.).

class ResearcherAgent(FilePipelineAgent):
    def execute_llm_task(self, metadata: dict, body: str) -> tuple[dict, str]:
        print(f"🔍 [{self.name}] Gathering data for: '{body}'...")
        
        # Simulating LLM response
        llm_research_notes = f"# Research for: {body}\n\n- Fact 1: Found key data.\n- Fact 2: Verified sources."
        metadata["researcher_tokens"] = 450 # tracking metrics
        
        return metadata, llm_research_notes

class WriterAgent(FilePipelineAgent):
    def execute_llm_task(self, metadata: dict, body: str) -> tuple[dict, str]:
        print(f"✍️ [{self.name}] Drafting article based on research...")
        
        # Simulating LLM response
        llm_draft = f"{body}\n\n## Introduction\nThis is a fully drafted article based on the facts above."
        metadata["word_count"] = len(llm_draft.split())
        
        return metadata, llm_draft

🏃‍♂️ Putting it into Motion

You can orchestrate this with a simple file watcher or a master loop that triggers agents sequentially.

if __name__ == "__main__":
    # 1. Setup the Pipeline Stages
    stage_1 = "01_backlog"
    stage_2 = "02_researched"
    stage_3 = "03_drafted"
    
    # 2. Initialize Agents
    researcher = ResearcherAgent("Researcher-Alpha", stage_1, stage_2)
    writer = WriterAgent("Writer-Beta", stage_2, stage_3)
    
    # 3. Inject a starting task into the backlog folder
    os.makedirs(stage_1, exist_ok=True)
    with open(f"{stage_1}/quantum_computing.md", "w") as f:
        f.write("---\ntitle: Quantum Computing Basics\n---\nWrite a report on topological qubits.")
        
    print("--- 🚀 Starting Agentic Pipeline ---")
    
    # 4. Run the pipeline
    researcher.process_next_file() # Moves file from 01 -> 02
    writer.process_next_file()     # Moves file from 02 -> 03

💡 Best Practices for this Framework

⚠️ The File Locking Problem: If you scale this to run asynchronously (e.g., using a library like watchdog to trigger agents instantly when a file drops), ensure two agents don't try to open the same file at the exact same split-second. Moving a file (shutil.move or .unlink()) is generally atomic on modern OS filesystems, but adding a tiny .lock file or processing file-by-file sequentially avoids race conditions.

Idempotency: Make sure that if an agent crashes halfway through a task, it doesn't leave a corrupted or half-written markdown file in the destination folder. Always write to a temporary file (filename.md.tmp) and rename it to .md only when the write is 100% complete.
Git Integration: Since everything is Markdown, you can turn the root folder into a Git repository. You can have your orchestrator automatically commit changes after an agent finishes, giving you a beautiful, visual git diff of what each agent changed.

I would expect follow-on agents to search the specified folder for available (i.e. "waiting") markdown folders, and get their assignments in that method.

Got it—so instead of a master script pushing files through a rigid chain, the agents themselves are autonomous. They poll their assigned input folders, grab whatever is waiting for them, do the work, and stage it forward. This turns the folders into actual task queues.

Here is how you can refactor the framework to support this self-polling behavior, including handling file locks so two agents of the same type don't accidentally grab the same file if you scale up.

🛠️ The Autonomous Agent Implementation

We will add a .listen() method. This puts the agent into a loop where it scans its input folder, processes files as they arrive, and handles them safely.

import os
import time
from pathlib import Path
import yaml

class AutonomousAgent:
    def __init__(self, name: str, watch_dir: str, output_dir: str, poll_interval: int = 5):
        self.name = name
        self.watch_dir = Path(watch_dir)
        self.output_dir = Path(output_dir)
        self.poll_interval = poll_interval
        
        self.watch_dir.mkdir(parents=True, exist_ok=True)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def listen(self):
        """Puts the agent into an autonomous polling loop looking for work."""
        print(f"🤖 [{self.name}] Started. Watching folder: '{self.watch_dir.name}/'...")
        
        try:
            while True:
                # Find all markdown files, ignoring temporary or processing files
                available_tasks = [
                    f for f in self.watch_dir.glob("*.md") 
                    if not f.name.startswith(".") and not f.name.endswith(".tmp")
                ]
                
                if available_tasks:
                    print(f"🎯 [{self.name}] Found {len(available_tasks)} task(s) waiting.")
                    for task_file in available_tasks:
                        self._claim_and_process(task_file)
                
                time.sleep(self.poll_interval)
        except KeyboardInterrupt:
            print(f"\n🛑 [{self.name}] Shutting down gracefully.")

    def _claim_and_process(self, file_path: Path):
        """Claims the file to prevent race conditions, processes it, and stages it forward."""
        # 1. Claim the file by renaming it to a hidden processing state
        processing_path = file_path.with_name(f".processing_{self.name}_{file_path.name}")
        try:
            file_path.rename(processing_path)
        except FileNotFoundError:
            # Another agent grabbed it first!
            return

        print(f"💼 [{self.name}] Claimed task: {file_path.name}")

        try:
            # 2. Read and parse
            metadata, body = self.parse_markdown(processing_path)
            
            # 3. Run the LLM/Agent assignment logic
            updated_metadata, updated_body = self.execute_assignment(metadata, body)
            
            # 4. Save to next folder safely (Atomic Write)
            destination_path = self.output_dir / file_path.name
            temp_dest_path = destination_path.with_suffix('.md.tmp')
            
            # Write frontmatter + content
            if "history" not in updated_metadata:
                updated_metadata["history"] = []
            updated_metadata["history"].append(f"Completed by {self.name}")
            
            frontmatter = yaml.dump(updated_metadata, default_flow_style=False)
            with open(temp_dest_path, 'w', encoding='utf-8') as f:
                f.write(f"---\n{frontmatter}---\n\n{updated_body}")
            
            # Final atomic move into the destination folder
            temp_dest_path.rename(destination_path)
            print(f"🚀 [{self.name}] Task staged forward to '{self.output_dir.name}/'")
            
            # 5. Clean up the original file from the input folder
            processing_path.unlink()

        except Exception as e:
            print(f"❌ [{self.name}] Error processing {file_path.name}: {e}")
            # Rollback: rename it back so it's "waiting" again for a retry
            if processing_path.exists():
                processing_path.rename(file_path)

    def parse_markdown(self, file_path: Path) -> tuple[dict, str]:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        if content.startswith('---'):
            parts = content.split('---', 2)
            if len(parts) >= 3:
                return yaml.safe_load(parts[1]) or {}, parts[2].strip()
        return {}, content.strip()

    def execute_assignment(self, metadata: dict, body: str) -> tuple[dict, str]:
        """Override this method with actual agent logic."""
        raise NotImplementedError

🏃‍♂️ Running the Autonomous Pipeline

To see this in action, you would run your agents as separate background processes or threads. Here is how a specific Editor Agent would look and run:

class EditorAgent(AutonomousAgent):
    def execute_assignment(self, metadata: dict, body: str) -> tuple[dict, str]:
        print(f"✍️ [{self.name}] Polishing and formatting the markdown...")
        
        # Simulating LLM assignment work
        polished_body = f"{body}\n\n## Editor's Note\nThis document was reviewed and approved."
        metadata["status"] = "APPROVED"
        metadata["editor_notes"] = "Grammar fixed, layout adjusted."
        
        return metadata, polished_body

if __name__ == "__main__":
    # This agent spins up and just sits there, watching '03_drafted' 
    # and dropping finished work into '04_reviewed'
    editor = EditorAgent(
        name="Editor-Agent-1", 
        watch_dir="03_drafted", 
        output_dir="04_reviewed",
        poll_interval=2
    )
    
    editor.listen()

💎 Why this Pattern is Robust

Atomic File Operations: By renaming the file to .processing_AgentName_filename.md instantly, we "lock" the file. If you run three instances of the same Editor Agent to handle heavy load, they won't fight over the same file.
Crash Resilience: If an agent's LLM call crashes midway through, the except block catches it and renames the file back to its original name. The task goes back into the "waiting" pool for the next available agent.
Infinite Scaling: You can spin this script up on completely different servers. As long as they point to the same network file share (like NFS, AWS EFS, or a shared Docker volume), they will pull assignments dynamically.

Think innovatively. How might blockchain fit into this framework?

When an enterprise framework hits an intersection of highly complex dependencies, multi-stakeholder execution, and absolute traceability, dropping a traditional database for a ledger can be a massive paradigm shift.

Looking at your campus IT framework through the lens of systematic innovation (like TRIZ or First Principles), we shouldn’t just look at blockchain as a trendy way to store a table. Instead, we look at it to solve the fundamental friction points of trust, state-tracking, and automated execution.

Here is how a blockchain layer transforms this scheduling framework from a localized Python script into an un-tamperable, decentralized operations engine.

1. From "CSV Templates" to Smart Contracts

Right now, your Python script reads a CSV file to check dependencies and shift windows. In a blockchain-driven architecture, the Global Rules (Table 3) and Task Dependencies are written directly as Smart Contracts (e.g., Solidity on a private EVM network like Hyperledger Besu or an enterprise L2).

The Innovation: A task is no longer a row in a spreadsheet; it is an On-Chain Cryptographic Object.
How it Works: Task B (W-201) literally cannot accept a technician assignment until the smart contract validates that Task A (D-101) has posted a "Completed" transaction hash to the ledger. The timeline enforces itself natively at the protocol layer, completely removing the risk of a human manually checking off a dependency prematurely.

2. Immutable SQA Auditing (The "Provenance" Layer)

For environments where process compliance and safety assurance are paramount, knowing exactly who did what and when is a hard requirement.

The Grid: Instead of a local CSV output that can be edited or deleted, every schedule change, task start, transit buffer entry, and completion sign-off is a signed cryptographic transaction.
The Audit Trail: This creates an unbreakable chain of custody. If a core switch upgrade fails at 03:00 AM, the forensic team can look at the immutable block history to verify if the 20-minute transit buffer was actually taken, which specific technician's private key signed off on the prerequisite backup verification, and exactly how long the passive lag phase lasted. It turns the schedule into a live, mathematical audit trail.

3. Decentralized Multi-Tenant Scheduling

Right now, the framework assumes one central dispatcher or script rules all. But what happens if your 56 "Skill Gap" tasks are outsourced to external contractors, or if parts of the campus infrastructure are managed by third-party facilities vendors?

The Friction: External vendors don't want to use your internal software, and you don't want them accessing your internal active directory.
The Blockchain Solution: A shared ledger allows your internal IT team, third-party network contractors, and facilities management to interact with the exact same schedule without anyone owning the underlying infrastructure.
Tokenized Resource Allocation: Technicians could have their certified skills issued as Verifiable Credentials (VCs) or Soulbound Tokens (SBTs). When your Python solver looks for a qualified tech, it checks the ledger for an active cryptographic credential (e.g., Certified Cisco Senior Architect). If the external contractor has the token, they can autonomously claim the task via a smart contract escrow.

Architecture Blueprint: The Web3 Solver Bridge

   [ Data Ingestion ]       ->   External Vendors / Internal Teams (Sign with Private Keys)
          │
          ▼
   [ Smart Contracts ]      ->   Enforces Shift Windows, Skill Matches, & Dependencies Natively
          │
          ▼
   [ Local Python Solver ]  ->   Pulls current state from Ledger, uses CP-SAT locally on your 
          │                      hardware to crunch the math, and returns the optimized plan.
          ▼
   [ Block Committal ]      ->   Pushes the finalized execution path back to the chain as immutable "Work Orders."

4. Decentralized Orchestration of Local Compute (DePIN)

If you scale this framework to monitor thousands of interconnected infrastructure points across multiple facility domains, the optimization math gets incredibly heavy.

Instead of running a single script on a local machine, you could use a DePIN (Decentralized Physical Infrastructure Network) model. You could utilize the local hardware stacks across your environment (like your high-performance mini-PCs and local GPUs) to compute different slices of the optimization graph simultaneously. The node that discovers the most mathematically optimal schedule (the lowest objective score) submits the proof to the chain and is rewarded with validation priority.

The Verdict: Is it practical?

For simple, localized 100-task scheduling: No. A relational database or your optimized pandas/OR-Tools loop is cleaner and faster.
For cross-organizational, mission-critical operations where safety, process compliance, and verified accountability are non-negotiable: Yes. It replaces a fragile chain of emails and spreadsheets with a single, self-executing source of truth.

If we were to map this out using a systems thinking approach, which area of your operational framework faces the most real-world friction: process compliance reporting, or coordinating with external/unassigned resources?

From <https://gemini.google.com/app/318447f944c70e02> Google Gemini (3.5 Flash)

Search This Blog

Learning from ChatGPT